package com.example.mq.rocketmq.producer;

import com.example.mq.rocketmq.dataobject.Mq01Message;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * 使用springboot rocketMQTemplate 发送消息
 *
 * 我们只要把tags使用":"添加到topic后面就可以了。例如：xxxx:tag1 || tag2 || tag3
 *
 * @author luowj12
 * @since 2021/4/25 15:17
 */
@Component
public class RocketMQProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步发送
     * @param id
     * @return
     */
    public SendResult syncSend(Integer id) {
        // 创建 Mq01Message 消息
        Mq01Message message = new Mq01Message();
        message.setId(id);
        // 同步发送消息
        return rocketMQTemplate.syncSend(Mq01Message.TOPIC, message);

    }

    /**
     * 异步有回调的发送
     * @param id
     */
    public void asyncSend(Integer id) {
        // 创建 Mq01Message 消息
        Mq01Message message = new Mq01Message();
        message.setId(id);
        // 异步发送消息
        rocketMQTemplate.asyncSend(Mq01Message.TOPIC, message, new SendCallback() {

            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }
        });
    }

    /**
     * 同步顺序消费
     * @param id
     * @return
     */
    public SendResult syncSendOrderly(Integer id) {
        // 创建 Demo06Message 消息
        Mq01Message message = new Mq01Message();
        message.setId(id);
        int hash = id%3;
        // 同步发送消息
        return rocketMQTemplate.syncSendOrderly(Mq01Message.TOPIC, message, String.valueOf(hash));
    }

    /**
     * 异步顺序消费
     * @param id
     * @param callback
     */
    public void asyncSendOrderly(Integer id, SendCallback callback) {
        // 创建 Demo06Message 消息
        Mq01Message message = new Mq01Message();
        message.setId(id);
        // 异步发送消息
        rocketMQTemplate.asyncSendOrderly(Mq01Message.TOPIC, message, String.valueOf(id), callback);
    }



    /**
     * oneway 发送
     * @param id
     */
    public void onewaySend(Integer id) {
        // 创建 Mq01Message 消息
        Mq01Message message = new Mq01Message();
        message.setId(id);
        // oneway 发送消息
        rocketMQTemplate.sendOneWay(Mq01Message.TOPIC, message);
    }

}
